package com.event.ingestion.kafka;

import com.event.ingestion.IngestionExecutor;
import com.event.ingestion.config.LoadConfig;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KStreamBuilder;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

/**
 * @author ZzHh
 * @Classname BaseStreamer
 * @Description TODO
 * @Date 2020/6/17 10:44
 * @Create by IntelliJ IDEA
 **/

/**
 * kafkaStream数据变换接口定义
 */
public abstract class BaseStreamer implements IngestionExecutor {
    //定义参数常量
    private String zookeeperUrl = null;

    private String kafkaBrokerUrl = null;

    //数据变换过程中产生的中间数据
    private String stateDir = null;

    //指定程序运行的ID
    protected abstract String getApplicationId();

    //数据源
    protected abstract String getSourceTopic();

    //目的地
    protected abstract String getTargetTopic();

    //初始化配置参数
    public void initialize(Properties properties) {
        this.zookeeperUrl = properties.getProperty(LoadConfig.zookeeperUrl);
        this.kafkaBrokerUrl = properties.getProperty(LoadConfig.kafkaBrokerUrl);
        this.stateDir = properties.getProperty(LoadConfig.stateDir);
    }

    //定义子类实现的数据业务拆分方法
    //(12312321,[213,123,123,123])
    //([12312321,1213])
    //([12312321,5324])
    protected abstract List<String[]> transform(String [] fields);

    //子类校验头信息
    protected abstract Boolean isHeader(String [] fields);

    //子类校验记录是否有效
    protected abstract Boolean isValid(String [] fields);

    //业务代码执行入口
    protected void stream() throws Exception {
        //定义Properties封装初始化参数
        Properties properties = new Properties();

        //设置自动偏移量,从头开始
        //设置便宜复位到最早,可用相同的预加载数据重新运行代码
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        //app_id
        properties.put(StreamsConfig.APPLICATION_ID_CONFIG,getApplicationId());
        //手动设置ApplicationId,解决二次执行KafkaStream数据不进入topic
        //properties.put(StreamsConfig.APPLICATION_ID_CONFIG,"BaseStreamer");

        //kafka地址
        properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,kafkaBrokerUrl);

        //zookeeperd地址
        properties.put(StreamsConfig.ZOOKEEPER_CONNECT_CONFIG,zookeeperUrl);

        //key-value的序列化和反序列化实例
        properties.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        properties.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());

        //设置state目录存放kafkaStream传输的状态信息
        properties.put(StreamsConfig.STATE_DIR_CONFIG,stateDir);

        //创建KStreamBuilder对象
        KStreamBuilder kStreamBuilder = new KStreamBuilder();

        //绑定sourceTopic
        KStream<String, String> stream = kStreamBuilder.stream(Serdes.String(), Serdes.String(), this.getSourceTopic());

        //数据变换
        KStream<String, String> result = stream.flatMap((k, v) -> transform(k, v)).filter((k, v) -> v != null && v.length() > 0);
        result.to(Serdes.String(),Serdes.String(),this.getTargetTopic());

        //启动数据传输进程
        StreamsConfig config = new StreamsConfig(properties);
        (new KafkaStreams(kStreamBuilder,config)).start();

    }

    //数据转换方法
    private Iterable<KeyValue<String,String>> transform(String key,String value) {
        List<KeyValue<String,String>> items = new ArrayList<KeyValue<String,String>>();
        //对value进行切割
        String [] fields = value.split(",", -1);
        if (isHeader(fields) || !isValid(fields)) {
            //不满足要求的将value设置为空字符串
            items.add(new KeyValue<>(key,""));
        }else{
            for (String[] vs :transform(fields)) {
                items.add(new KeyValue<>(key,String.join(",",vs)));
            }
        }
        return items;
    }

    //程序入口
    @Override
    public void executor(String[] args) throws Exception {
        //校验
        if (args.length < 1) {
            System.out.println("参数异常!");
        }else{
            try {
                this.initialize(LoadConfig.loadSettings(args[0]));
                this.stream();
            }catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

}